package cn.edu.laud.sany.dataimport;

import java.io.File;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import cn.edu.laud.sany.types.UnConfiguredCFException;
import cn.edu.laud.sany.utils.MutationMap;
import cn.edu.laud.sany.utils.MyLog;
import cn.edu.laud.sany.utils.ProcessManager;

public class LaUDRunner extends Thread
{
	// constants
	protected static ConsistencyLevel CL = ConsistencyLevel.ANY;
	
	// for import process
	protected int threadNum;
	protected String keyspace;
	protected MyLog logger;
	
	// for thrift connection
	protected String ip;
	protected TTransport tf;
	protected TProtocol proto;
	protected Cassandra.Client client;
	protected ConcurrentLinkedQueue<MutationMap> mutationMapQueue;
	
	public LaUDRunner(int no, String keyspace, ConcurrentLinkedQueue<MutationMap> mutationMapQueue)
	{
		this.threadNum = no;
		this.keyspace = keyspace;
		this.mutationMapQueue = mutationMapQueue;
		File foler = new File("log");
		if (foler.exists() == false)
		{
			foler.mkdir();
		}
		logger = MyLog.getMyLog(LaUDImport.getLogFilePath(), true);
		logger.setThreadName("Thread " + no);
		ip = LaUDImport.IP[threadNum % LaUDImport.IP.length];
	}
	
	private boolean openConnection()
	{
		try
		{
			logger.info("Try get Thrift connection...");
			tf = new TFramedTransport(new TSocket(ip, LaUDImport.laudPort));
			proto = new TBinaryProtocol(tf);
			client = new Cassandra.Client(proto);
			tf.open();
			client.set_keyspace(keyspace);
			logger.info("Get Thrift connection successful.");
		}
		catch (Exception e)
		{
			logger.error("Get Thrift connection error LaUD IP:" + ip);
			logger.error("Caused by: " + e.toString());
			return false;
		}
		return true;
	}
	
	private void closeConnection()
	{
		try
		{
			if (tf != null)
			{
				tf.close();
			}
			logger.info("Laud connect closed successfully.");
		}
		catch (Exception e)
		{
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}
	
	private void batchImport(Cassandra.Client nowClient, MutationMap map)
			throws UnConfiguredCFException
	{
		int exceptionCount = 0;
		boolean flag = false;
		while (flag == false)
		{
			try
			{
				nowClient.batch_mutate(map.getMap(), CL);
				ProcessManager.addRealImportCount(map.mutationCount);
				flag = true;
				map.clearMutation();
			}
			catch (InvalidRequestException e)
			{
				if (e.getWhy() != null && e.getWhy().startsWith("unconfigured columnfamily"))
				{
					logger.error("One batch Import error!");
					logger.error("Caused by: " + e.toString());
					logger.error("Please check you cfs file!");
					throw (new UnConfiguredCFException());
				}
				else
				{
					logger.error("One batch import error, will try again after 5 sec!");
					logger.error("Caused by: " + e.toString());
					WaitSecond(5);
				}
			}
			catch (UnavailableException e)
			{
				exceptionCount++;
				logger.error("One batch import error, will try again after 5 sec!");
				logger.error("Caused by: " + e.toString());
				WaitSecond(5);
			}
			catch (TimedOutException e)
			{
				exceptionCount++;
				logger.error("One batch Import time out, will try again after 5 sec !");
				WaitSecond(5);
			}
			catch (TTransportException e)
			{
				exceptionCount++;
				logger.error("One batch import error, will try to get connection after 5 sec!");
				logger.error("Caused by: " + e.toString());
				WaitSecond(5);
			}
			catch (TException e)
			{
				exceptionCount++;
				logger.error("One batch import error, will try again after 5 sec!");
				logger.error("Caused by: " + e.toString());
				WaitSecond(5);
			}
			if (exceptionCount > 5)
			{
				logger.error("Try to reconnect server...");
				while (openConnection() == false)
				{
					logger.info("Reconnect server failed, will wait 10 sec to retry...");
					WaitSecond(10);
				}
				exceptionCount = 0;
			}
		}
		
	}
	
	private void WaitSecond(int secound)
	{
		try
		{
			Thread.sleep(secound * 1000);
		}
		catch (InterruptedException e)
		{
		}
	}
	
	@Override
	public void run()
	{
		logger.info("Thread Start...");
		try
		{
			while (openConnection() == false)
			{
				logger.info("Begin import failed, will wait 3 sec to restart...");
				WaitSecond(3);
			}
			
			while (LaUDImport.halted == false)
			{
				MutationMap mutationMap = null;
				synchronized (mutationMapQueue)
				{
					logger.info("Queue:" + mutationMapQueue.size());
					mutationMap = mutationMapQueue.poll();					
				}
				if (mutationMap == null)
				{
					if (LaUDImport.nomoreCsv && LaUDImport.doNotWait)
					{
						logger.info("No more available csv/gz source file, thread will end.");
						break;
					}
					WaitSecond(1);
					continue;
				}
				// import
				batchImport(client, mutationMap);
			}
		}
		catch (Exception e)
		{
			logger.error(e.getMessage());
			e.printStackTrace();
		}
		finally
		{
			closeConnection();
		}
	}
}
